package aspirecn.project3.consumer;

import aspirecn.project3.feign.GoodServiceFeign;
import aspirecn.project3.request.MockData;
import aspirecn.project3.util.JsonUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;
/**
 * packageName aspirecn.project3.service
 *
 * @author 邵一宸
 * @version JDK 8
 * @className KafkaConsumerService
 * @date 2024/7/20
 * @description TODO
 */


@Slf4j
@Component
public class KafkaConsumerService {

    @Autowired
    GoodServiceFeign goodServiceFeign;

    @KafkaListener(groupId = "mockGroup", topics = "mock_data_topic")
    public void onMessage(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
        log.info("消费端接收消息: {}", record.value());
//        JsonUtil.json2Obj(record.value(),MockData.class);

        JSONObject jsonObject = JSON.parseObject((String)record.value());
        MockData data = JSON.toJavaObject(jsonObject, MockData.class);
//        System.out.println(data);
        goodServiceFeign.addGoodPointCountsById(data.getGoodId(), data.getPointCounts());
        //	收工签收机制
        acknowledgment.acknowledge();
    }


}